package cn.lsh.spark;

import org.apache.commons.lang.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import scala.Tuple2;

import java.util.*;

public class RDDTest {

	public static void main(String[] args) throws Exception {
		SparkConf conf = new SparkConf();
		//master=spark://node00:7077,node02:7077 提交到spark自带的standalone集群
		// conf.setMaster("spark://node00:7077,node02:7077");

		//local后面指定数字，可以在本地多线程运行，多个task也就有多个分区，不知道这里设置为3，也是两个分区，难道最大只有2个？？？
		conf.setMaster("local[3]");
		conf.setAppName("test");
		//client连接standlaone集群需要指定jar包路径
		// conf.setJars(new String[] { "G:\\bigdata\\hadoop-test\\target\\hadoop-test-1.0-SNAPSHOT.jar" });

		JavaSparkContext sparkContext = new JavaSparkContext(conf);

		// testSortByKey(sparkContext);
		// testCache(sparkContext);
		// testCheckPoint(sparkContext);
		// testSample(sparkContext);
		// testTakeSample(sparkContext);
		// testTake(sparkContext);
		// testAggregate(sparkContext);
		testAggregateByKey(sparkContext);
		// testJoin(sparkContext);
		// testUnion(sparkContext);
		// testCogroup(sparkContext);
		// testMapPartition(sparkContext);
		// testMapPartitionWithIndex(sparkContext);
		// testZip(sparkContext);
		// testAction(sparkContext);
		// testFlatMap(sparkContext);
		// testGroupBy(sparkContext);

		sparkContext.stop();
	}

	public static void testFilter(JavaSparkContext sparkContext) {
		JavaRDD<Integer> parallelize = sparkContext.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6));
		//注意，spark会将在Executor中使用到的black变量发送到对应的执行task，这样有多少个task就有多少个副本变量
		List<Integer> black = Arrays.asList(1, 2, 3);
		//可以通过broadcast()方法广播变量，这样一个Executor只有一个变量副本。但是不能广播RDD
		Broadcast<List<Integer>> broadcast = sparkContext.broadcast(black);
		parallelize.filter(broadcast.getValue()::contains).foreach(i -> System.out.println(i));
	}

	public static void testSortByKey(JavaSparkContext sparkContext) {
		JavaRDD<String> lines = sparkContext.textFile("file:/bigdata/hadoop-test/input/spark/scores.txt");
		JavaPairRDD<Scores.BjScore, String> jpRddRDD = lines.mapToPair(line -> new Tuple2<>(new Scores.BjScore(line.split(" ")[0], Integer.parseInt(line.split(" ")[1])), ""));
		System.out.println("分区数：" + jpRddRDD.getNumPartitions());
		JavaPairRDD<String, Integer> stringIntegerRDD = jpRddRDD.sortByKey(false).mapToPair(o ->
		{
			Tuple2<String, Integer> t = new Tuple2<>(o._1().getBj(), o._1().getScore());
			System.out.println("mapToPair----" + t);
			return t;
		});
		stringIntegerRDD.cache();
		//foreach是流式输出，体现不出sortByKey的排序
		stringIntegerRDD.foreach(o -> System.out.println(o));
		System.out.println("--------------------------------collect after forEach");
		//先collect后输出的结果是正确的全排序
		stringIntegerRDD.collect().forEach(System.out::println);
	}

	/**
	 * 缓存算子，有三种算子
	 * 一、cache：只放在内存，如果数据量大于内存，则不会进行缓存
	 * 二、persist：可以选择缓存类型，主要有四个参数控制，通常使用内存加磁盘，当内存用满后，会将数据缓存到磁盘
	 * 1、useDisk：使用磁盘
	 * 2、useMemory：使用内存
	 * 3、useOffHeap：使用对外内存
	 * 4、deserialized：反序列化
	 * 5、replication：保存副本数
	 * 三、checkpoint：将RDD数据放在磁盘中，需要指定目录，且job运行完后数据依然存在，不会被清除
	 *
	 * @param sparkContext
	 */
	public static void testCache(JavaSparkContext sparkContext) {
		JavaRDD<String> linesRdd = sparkContext.textFile("hdfs://mycluster/test/input/firend/friden_list.txt");

		//缓存RDD
		linesRdd.cache();
		//persist，可以选择缓存方式，persist() = persist(StorageLevel.MEMORY_ONLY()) = cache()
		// linesRdd.persist(StorageLevel.MEMORY_ONLY());

		long startTime1 = System.currentTimeMillis();
		long result1 = linesRdd.count();
		System.out.println("count1: " + result1 + ", time: " + (System.currentTimeMillis() - startTime1));

		long startTime2 = System.currentTimeMillis();
		long result2 = linesRdd.count();
		System.out.println("count2: " + result2 + ", time: " + (System.currentTimeMillis() - startTime2));
	}

	/**
	 * 设置检查点，spark会将检查点算子的结果数据保存到磁盘，并切断和父RDD的依赖关系
	 *
	 * @param sparkContext
	 */
	public static void testCheckPoint(JavaSparkContext sparkContext) {
		//需要先设置数据存放目录
		sparkContext.setCheckpointDir("hdfs://mycluster/test/spark/checkpiont/friend");
		JavaRDD<String> lines = sparkContext.textFile("hdfs://mycluster/test/input/firend/friden_list.txt");
		lines.checkpoint();
		lines.count();
	}

	/**
	 * 抽样算子
	 * 1、参数一：每次抽取的数据是否放回
	 * 2、参数二：小数，抽样数据占总数的百分比，最终抽样的数据量是一个大概值
	 * 3、参数三：种子值，当前两个参数相同时相同的种子值每次抽样的结果一致
	 *
	 * @param sparkContext
	 */
	public static void testSample(JavaSparkContext sparkContext) {
		JavaRDD<String> linesRdd = sparkContext.textFile("hdfs://mycluster/test/input/weather/tq_data.txt");

		//不能简写成System.out::println
		linesRdd.sample(false, 0.0001, 100000).foreach(s -> System.out.println(s));
	}

	/**
	 * 和sample一样是抽样算子
	 * 1、参数一：每次抽取的数据是否放回
	 * 2、参数二：整数，抽样数据量
	 * 3、参数三：种子值，当前两个参数相同时相同的种子值每次抽样的结果一致
	 *
	 * @param sparkContext
	 */
	public static void testTakeSample(JavaSparkContext sparkContext) {
		JavaRDD<String> linesRdd = sparkContext.textFile("hdfs://mycluster/test/input/weather/tq_data.txt");
		//注意和sample不一样，takeSample返回不是RDD，而是一个List
		linesRdd.takeSample(false, 10).forEach(System.out::println);
	}

	/**
	 * take，action算子，获取RDD中的前n条数据
	 *
	 * @param sparkContext
	 */
	public static void testTake(JavaSparkContext sparkContext) {
		JavaRDD<String> linesRdd = sparkContext.textFile("hdfs://mycluster/test/input/weather/tq_data.txt");
		linesRdd.take(10).forEach(System.out::println);
	}

	/**
	 * 聚合函数，和reduce类似，但reduce返回值类型不能变，aggregate可以指定聚合后的返回值类型
	 * 1、参数一：返回类型的初始值
	 * 2、参数二：累加器，第一个参数为上一次的入参，第二个参数为当前处理的数据，返回累加后的结果
	 * 3、参数三：合并器，合并累加器的结果,因为每个节点是在本地进行累加的
	 *
	 * @param sparkContext
	 */
	public static void testAggregate(JavaSparkContext sparkContext) {
		JavaRDD<String> linesRdd = sparkContext.textFile("hdfs://mycluster/test/input/weather/tq_data.txt", 3);
		int total = linesRdd.aggregate(Integer.MIN_VALUE, (t1, t2) -> {
			String s = StringUtils.split(t2, ' ')[2];
			return t1 + Integer.parseInt(s.substring(0, s.length() - 1));
		}, Integer::sum);
		System.out.println(total);
	}

	/**
	 * aggregateByKey相当于是aggregate和groupByKey函数组合
	 * 1、第一个参数为value聚合的初始值
	 * 2、第二个参数是各个分区内相同key的聚合函数
	 * 3、第三个参数是分区合并后相同key的聚合函数，在集群后本地多线程的情况下可以体现出来
	 *
	 * @param sparkContext
	 */
	public static void testAggregateByKey(JavaSparkContext sparkContext) {
		List<Tuple2<String, Integer>> list = Arrays.asList(
				new Tuple2<>("math", 100),
				new Tuple2<>("hadoop", 80),
				new Tuple2<>("hadoop", 52),
				new Tuple2<>("math", 50),
				new Tuple2<>("sic", 79),
				new Tuple2<>("sic", 46));
		JavaPairRDD<String, Integer> parallelizePairs = sparkContext.parallelizePairs(list);
		JavaPairRDD<String, Integer> result = parallelizePairs.aggregateByKey(0, (last, next) -> {
			System.out.println("seqFunc: " + last + " + " + next);
			return last + next;
		}, (s1, s2) -> {
			System.out.println("comFunc: " + s1 + " + " + s2);
			return s1 + s2;
		});
		System.out.println("***************Result**************");
		result.foreach(o -> System.out.println(o));
	}

	/**
	 * 对应表的连接方式，连接两个二元组RDD的key的类型必须一致，返回值都是JavaPairRDD<String, Tuple2<Optional<String>, Optional<Integer>>>这种类型，
	 * RDD存的是一个复合Tuple2，外层Tuple2的key是左边RDD的key，内层Tuple2的key是左边RDD的value，内层Tuple2的value是右边RDD的value
	 * 1、join：返回类型JavaPairRDD<String, Tuple2<String, Integer>>
	 * 2、leftOuterJoin：返回类型JavaPairRDD<String, Tuple2<String, Optional<Integer>>>，右边的RDD可能为空
	 * 3、rightOuterJoin：返回类型JavaPairRDD<String, Tuple2<Optional<String>, Integer>>，左边的RDD可能为空
	 * 4、fullOuterJoin：返回类型JavaPairRDD<String, Tuple2<Optional<String>, Optional<Integer>>>，都可能为空
	 *
	 * @param sparkContext
	 */
	public static void testJoin(JavaSparkContext sparkContext) {
		List<Tuple2<String, String>> list = new ArrayList<>();
		list.add(Tuple2.apply("hadoop", "hdfs"));
		list.add(Tuple2.apply("hadoop", "yarn"));
		list.add(Tuple2.apply("scala", "spark"));
		list.add(Tuple2.apply("hbase", "hive"));
		list.add(Tuple2.apply("spark", "standalone"));
		JavaPairRDD<String, String> rdd1 = sparkContext.parallelizePairs(list);
		JavaPairRDD<String, Integer> rdd2 = sparkContext.parallelizePairs(Arrays.asList(
				Tuple2.apply("hadoop", 1), Tuple2.apply("hbase", 2), Tuple2.apply("spark", 3), Tuple2.apply("java", 4)
		));
		JavaPairRDD<String, Tuple2<String, Integer>> join = rdd1.join(rdd2);
		System.out.println("******************内连接：只保留两个RDD中key相同的数据");
		join.foreach(o -> System.out.println(o));

		System.out.println("******************左外连接：保留左边RDD的全部数据，不在右边RDD中的Tuple2中的value为空");
		JavaPairRDD<String, Tuple2<String, Optional<Integer>>> leftJoin = rdd1.leftOuterJoin(rdd2);
		leftJoin.foreach(o -> System.out.println(o));

		System.out.println("******************右外连接：保留右边RDD的全部数据，不在左边RDD中的Tuple2中的key为空");
		JavaPairRDD<String, Tuple2<Optional<String>, Integer>> rightJoin = rdd1.rightOuterJoin(rdd2);
		rightJoin.foreach(o -> System.out.println(o));

		System.out.println("******************全连接（左外连接 + 右外连接）*****************");
		JavaPairRDD<String, Tuple2<Optional<String>, Optional<Integer>>> fullJoin = rdd1.fullOuterJoin(rdd2);
		fullJoin.foreach(o -> System.out.println(o));
	}

	/**
	 * 集合操作算子
	 *
	 * @param sparkContext
	 */
	public static void testUnion(JavaSparkContext sparkContext) {
		JavaRDD<Integer> rdd1 = sparkContext.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6));
		JavaRDD<Integer> rdd2 = sparkContext.parallelize(Arrays.asList(4, 5, 6, 7, 8, 9));
		System.out.println("******************union：取并集");
		rdd1.union(rdd2).foreach(o -> System.out.println(o));
		System.out.println("******************intersection：取交集");
		rdd1.intersection(rdd2).foreach(o -> System.out.println(o));
		System.out.println("******************subtract：取rdd1中不包含在rdd2中的元素集合");
		rdd1.subtract(rdd2).foreach(o -> System.out.println(o));
	}

	/**
	 * mapPartition可以指定分区数，有几个分区就调用几次回调函数，回调函数每次处理分区的所有数据
	 * foreachPartition：回调函数入参为所有数据，可以自己实现迭代遍历
	 *
	 * @param sparkContext
	 */
	public static void testMapPartition(JavaSparkContext sparkContext) {
		JavaRDD<Integer> rdd = sparkContext.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6), 3);
		rdd.mapPartitions((FlatMapFunction<Iterator<Integer>, Integer>) t -> {
			List<Integer> list = new ArrayList<>();
			System.out.println("创建数据库连接。。。");
			while (t.hasNext()) {
				System.out.println("插入数据。。。");
				list.add(t.next() + 1);
			}
			System.out.println("关闭数据库连接。。。");
			return list.iterator();
		}).foreach(o -> System.out.println(o));
		rdd.foreachPartition(f -> {
			while (f.hasNext()) {
				System.out.println(f.next());
			}
		});
	}

	/**
	 * 带分区下标的mapPartition
	 * 1、第一个参数：匿名函数，匿名函数第二个入参和mapPartition一致，第一个入参为partition下标
	 * 2、第二个参数：是否继承上一个RDD的partition数
	 *
	 * @param sparkContext
	 */
	public static void testMapPartitionWithIndex(JavaSparkContext sparkContext) {
		JavaRDD<Integer> rdd = sparkContext.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6), 3);
		JavaRDD<String> rdd1 = rdd.mapPartitionsWithIndex((index, t) -> {
			List<String> list = new ArrayList<>();
			while (t.hasNext()) {
				list.add("rdd1 partition：【" + index + "】, value：【" + t.next() + "】");
			}
			return list.iterator();
		}, true);
		//repartition：重置分区数，宽依赖，会产生shuffle落地磁盘
		// JavaRDD<String> rdd2 = rdd1.repartition(4);
		//coalesce也可以重置分区，但是可以控制是否产生shuffle,，当选择true时和repartition一样；但是如果分区数变大，shuffle指定false，不会改变分区数
		JavaRDD<String> rdd2 = rdd1.coalesce(4, false);
		rdd2.mapPartitionsWithIndex((index, t) -> {
			List<String> list = new ArrayList<>();
			while (t.hasNext()) {
				list.add("rdd2 partition：【" + index + "】, value：【" + t.next() + "】");
			}
			return list.iterator();
		}, true).foreach(o -> System.out.println(o));
	}

	/**
	 * zip：将两个RDD组合成一个K，V格式的RDD，这里两个RDD的分区数和元素个数都要相同
	 * zipWithIndex：将一个RDD变成成一个K，V格式的RDD，key为原RDD元素本身，value为元素下标，从0开始
	 * zipWithUniqueId：将一个RDD变成成一个K，V格式的RDD，并为每个元素生成一个新的唯一id，
	 * 这个id生成规则：：
	 * 1、每个分区中第一个元素的唯一ID值为：该分区索引号
	 * 2、每个分区中第N个元素的唯一ID值为：(前一个元素的唯一ID值) + (该RDD总的分区数)
	 * zipPartitions：要求分区必须一致，元素个数可以不一致，需要自己实现同分区内两个RDD元素的拼接
	 *
	 * @param sparkContext
	 */
	public static void testZip(JavaSparkContext sparkContext) {
		JavaRDD<Integer> rdd1 = sparkContext.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6), 3);
		JavaRDD<Integer> rdd2 = sparkContext.parallelize(Arrays.asList(6, 5, 4, 3, 2, 1), 3);
		JavaPairRDD<Integer, Integer> zip = rdd1.zip(rdd2);
		System.out.println("***********zip**********");
		zip.foreach(o -> System.out.println(o));
		System.out.println("***********zipWithIndex**********");
		rdd1.zipWithIndex().foreach(o -> System.out.println(o));
		System.out.println("***********zipWithUniqueId**********");
		rdd1.zipWithUniqueId().foreach(o -> System.out.println(o));

		System.out.println("***********zipPartitions**********");
		JavaRDD<Integer> rdd3 = sparkContext.parallelize(Arrays.asList(9, 8, 7, 6, 5, 4, 3, 2, 1), 3);
		JavaRDD<Integer> retRdd = rdd1.zipPartitions(rdd3, (t1, t2) -> {
			List<Integer> list = new ArrayList<>();
			while (t1.hasNext() && t2.hasNext()) {
				int i1 = t1.next();
				int i2 = t2.next();
				System.out.println(i1 + " + " + i2);
				list.add(i1 + i2);
			}
			while (t1.hasNext()) {
				list.add(t1.next());
			}
			while (t2.hasNext()) {
				list.add(t2.next());
			}
			return list.iterator();
		});
		retRdd.foreach(o -> System.out.println(o));
	}

	/**
	 * countByKey：根据二元组RDD的key计数，返回Map<Object, Long>
	 * countByValue：根据二元组RDD的key+value计数，返回Map<Tuple2<Object, Object>, Long>
	 *
	 * @param sparkContext
	 */
	public static void testAction(JavaSparkContext sparkContext) {
		JavaPairRDD<String, Integer> rdd1 = sparkContext.parallelizePairs(Arrays.asList(
				Tuple2.apply("hadoop", 1), Tuple2.apply("hadoop", 3), Tuple2.apply("hbase", 2),
				Tuple2.apply("spark", 3), Tuple2.apply("scala", 4), Tuple2.apply("spark", 3)
		));
		Map<String, Long> keyMap = rdd1.countByKey();
		System.out.println(keyMap);
		Map<Tuple2<String, Integer>, Long> valueMap = rdd1.countByValue();
		System.out.println(valueMap);
		System.out.println("********reduceByKey********");
		rdd1.reduceByKey(Integer::sum).foreach(o -> System.out.println(o));
		System.out.println("********distinct********");
		rdd1.distinct().foreach(o -> System.out.println(o));
	}

	/**
	 * flatMap：合并流，对集合中每个元素进行操作然后再扁平化
	 *
	 * @param sparkContext
	 */
	public static void testFlatMap(JavaSparkContext sparkContext) {
		JavaRDD<String> rdd1 = sparkContext.parallelize(Arrays.asList("123", "345", "456", "567", "789", "899"));
		JavaRDD<Integer> integerJavaRDD = rdd1.flatMap(o -> {
			List<Integer> list = new ArrayList<>();
			for (int i = 0; i < o.length(); i++) {
				list.add(Integer.valueOf(o.charAt(i) + ""));
			}
			return list.iterator();
		});
		integerJavaRDD.foreach(o -> System.out.println(o));
	}

	/**
	 * groupByKey：根据二元组key分组，返回二元组RDD中key还是原来的key，value是相同key的原来value的集合
	 * groupBy：自定义分组方式，入参为二元组，返回二元组RDD中key是用户指定的值，value是包括该值的原来二元组的集合
	 * groupWith：效果和cogroup类似，合并二元组RDD，把两个RDD相同key的value组合到一起
	 *
	 * @param sparkContext
	 */
	public static void testGroupBy(JavaSparkContext sparkContext) {
		JavaPairRDD<String, Integer> rdd1 = sparkContext.parallelizePairs(Arrays.asList(
				Tuple2.apply("hadoop", 1), Tuple2.apply("hadoop", 3), Tuple2.apply("hbase", 2),
				Tuple2.apply("spark", 3), Tuple2.apply("scala", 4), Tuple2.apply("spark", 3)
		));
		JavaPairRDD<String, Iterable<Integer>> rdd2 = rdd1.groupByKey();
		rdd2.foreach(o -> System.out.println(o));
		JavaPairRDD<Integer, Iterable<Tuple2<String, Integer>>> rdd3 = rdd1.groupBy(Tuple2::_2);
		rdd3.foreach(o -> System.out.println(o));
		System.out.println("**************************************");
		JavaPairRDD<String, Integer> rdd4 = sparkContext.parallelizePairs(Arrays.asList(
				Tuple2.apply("hadoop", 1), Tuple2.apply("hbase", 2), Tuple2.apply("hbase", 2), Tuple2.apply("spark", 3), Tuple2.apply("java", 4)
		));
		JavaPairRDD<String, Tuple2<Iterable<Integer>, Iterable<Integer>>> rdd5 = rdd1.groupWith(rdd4);
		rdd5.foreach(o -> System.out.println(o));
	}

	/**
	 * Cogroup：合并两个K，V格式RDD中相同key的Value
	 * 返回JavaPairRDD<String, Tuple2<Iterable<Integer>, Iterable<Integer>>>
	 * 内层二元组key为左边RDD中相同key的vlaue集合
	 * 内层二元组value为右边RDD中相同key的vlaue集合
	 *
	 * @param sparkContext
	 */
	public static void testCogroup(JavaSparkContext sparkContext) {
		JavaPairRDD<String, Integer> rdd1 = sparkContext.parallelizePairs(Arrays.asList(
				Tuple2.apply("hadoop", 1), Tuple2.apply("hadoop", 3), Tuple2.apply("hbase", 2),
				Tuple2.apply("spark", 3), Tuple2.apply("scala", 4), Tuple2.apply("spark", 3)
		));
		JavaPairRDD<String, Integer> rdd2 = sparkContext.parallelizePairs(Arrays.asList(
				Tuple2.apply("hadoop", 1), Tuple2.apply("hbase", 2), Tuple2.apply("hbase", 2), Tuple2.apply("spark", 3), Tuple2.apply("java", 4)
		));
		JavaPairRDD<String, Tuple2<Iterable<Integer>, Iterable<Integer>>> cogroup = rdd1.cogroup(rdd2);
		cogroup.foreach(o -> System.out.println(o));
	}

}
